package com.bw.yk09;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

public class test4 {
    public static void main(String[] args) throws Exception {
        //创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(new MyKafkaUtil().getKafkaConsumer("tms_ods","test4_group"));
        //筛选订单和订单明细数据
        SingleOutputStreamOperator<String> DwdBoundSortBeanDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_org_bound"));
        //对流中的数据类型进行转换
        SingleOutputStreamOperator<JSONObject> mapDS = DwdBoundSortBeanDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        try{

                            JSONObject jsonObject=JSONObject.parseObject(s);
                            String op=jsonObject.getString("type");
                            if(op.equals("insert"))
                            {
                                return jsonObject;
                            }else{
                                JSONObject dataJsonObject=jsonObject.getJSONObject("data");
                                if(dataJsonObject!=null&&dataJsonObject.getString("sort_time")!=null)
                                    return jsonObject;
                            }
                        }catch (Exception e)
                        {
                            e.printStackTrace();
                        }
                        return null;
                    }
                }
        );




        //64002
        OutputTag<DwdBoundSortBean> t64002 = new OutputTag<DwdBoundSortBean>("t64002"){};
        //64003
        OutputTag<DwdBoundSortBean> t64003 = new OutputTag<DwdBoundSortBean>("t64003"){};
        //64004
        OutputTag<DwdBoundSortBean> t64004 = new OutputTag<DwdBoundSortBean>("t64004"){};
        //64005
        OutputTag<DwdBoundSortBean> t64005 = new OutputTag<DwdBoundSortBean>("t64005"){};

        SingleOutputStreamOperator<DwdBoundSortBean> t64001DS = mapDS.process(new ProcessFunction<JSONObject, DwdBoundSortBean>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, DwdBoundSortBean>.Context ctx, Collector<DwdBoundSortBean> out) throws Exception {
                JSONObject data = value.getJSONObject("data");
                String status = data.getString("status");
                DwdBoundSortBean bean=JSON.parseObject(data.toJSONString(),DwdBoundSortBean.class);
                if ("64001".equals(status)) {
                    out.collect(bean);
                } else if ("64002".equals(status)) {
                    ctx.output(t64002, bean);
                } else if ("64003".equals(status)) {
                    ctx.output(t64003, bean);
                } else if ("64004".equals(status)) {
                    ctx.output(t64004, bean);
                } else if ("64005".equals(status)) {
                    ctx.output(t64005, bean);
                }
            }
        });
        t64001DS.print();

        t64001DS.map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64001"));
        t64001DS.getSideOutput(t64002).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64002"));
        t64001DS.getSideOutput(t64003).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64003"));
        t64001DS.getSideOutput(t64004).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64004"));
        t64001DS.getSideOutput(t64005).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64005"));

        env.execute();
    }
}
